// Description: 描述
package parallel

import "github.com/zeromicro/go-zero/core/mr"

func warpFn(index int, f func(int) error) func() error {
	return func() error {
		return f(index)
	}
}

func warpFnNoRet(index int, f func(int)) func() {
	return func() {
		f(index)
	}
}

func warpSafetyFn(index int, f func(int) (interface{}, error)) func() (interface{}, error) {
	return func() (interface{}, error) {
		return f(index)
	}
}

func finishSafetyFn(fns []func() (interface{}, error), revRet func(interface{})) error {
	number := len(fns)
	// 防止协程过多，与go-zero框架默认值保持一致
	if number > 16 {
		number = 16
	}

	return mr.MapReduceVoid(func(source chan<- interface{}) {
		for _, fn := range fns {
			source <- fn
		}
	}, func(item interface{}, writer mr.Writer[interface{}], cancel func(error)) {
		fn := item.(func() (interface{}, error))
		ret, err := fn()
		if err != nil {
			cancel(err)
		} else {
			writer.Write(ret)
		}
	}, func(pipe <-chan interface{}, cancel func(error)) {
		for ret := range pipe {
			revRet(ret)
		}
	}, mr.WithWorkers(number))
}
